package com.warrior.gatherserver.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.warrior.gathercommon.dto.GatherProDto;
import com.warrior.gathercommon.enums.ExecuteTypeEnum;
import com.warrior.gatherserver.dto.GataDataEsDto;
import com.warrior.gatherserver.dto.MetricEsDto;
import com.warrior.gatherserver.grpc.Messages;
import com.warrior.gatherserver.service.MessageService;
import com.warrior.gatherserver.util.DateUtils;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @ClassName com.warrior.gatherservice.service.impl.MessageServiceImpl.java
 * @Author xujian
 * @Description
 * @CreateTime 2020年12月16日 11:15:00
 */
@Slf4j
@Service
public class MessageServiceImpl implements MessageService {

    private static final String PROPER_CLASSS_PATH = "application.properties";

    private static final String PROPER_ENABLE = "enable";

    private static final String PROPER_ENABLE_TRUE = "true";

    private static final String PROPER_DELAY = "delay";

    private static final String PROPER_POINT = ".";

    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Value("${spring.elasticsearch.gather-index}")
    private String gatherIndex;

    @Override
    public List<GatherProDto> getProperties() {
        List<GatherProDto> propertys = new ArrayList<>();
        try {
            ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
            Resource[] resources = resolver.getResources(PROPER_CLASSS_PATH);
            for (Resource r : resources) {
                Properties properties = PropertiesLoaderUtils.loadProperties(r);
                Set<String> keys = properties.stringPropertyNames().stream().map(item -> item.split("\\.")[0]).collect(Collectors.toSet());
                keys.forEach(key -> {
                    String value = properties.getProperty(key + PROPER_POINT + PROPER_ENABLE);
                    if (StringUtils.isEmpty(value) || !PROPER_ENABLE_TRUE.equals(value)) {
                        return;
                    }

                    GatherProDto proDto = new GatherProDto();
                    proDto.setKey(key);
                    String delay = properties.getProperty(key + PROPER_POINT + PROPER_DELAY);
                    if(StringUtils.isEmpty(value)){
                        proDto.setType(ExecuteTypeEnum.ONE.getKey());
                    }else{
                        //TODO 待补充其他类型
                        proDto.setType(ExecuteTypeEnum.DELAY.getKey());
                        proDto.setDelay(Long.valueOf(delay));
                    }

                    log.info("name:{}, enable:{}, delay:{}", key, value, delay);
                    propertys.add(proDto);
                });
            }
        } catch (Exception e) {
            log.error("============ 获取待执行的任务配置异常 msg:{} =============", e.getMessage());
        }
        return propertys;
    }

    @Override
    public String handleMessage(Messages messages) {
        String result = null;
        try {
            GataDataEsDto esDto = new GataDataEsDto();
            esDto.setCreateTime(DateUtils.getCurrentDate(null, null));
            esDto.setCreateTimeStr(esDto.getCreateTime());
            esDto.setMetricType(messages.getMetricType());
            esDto.setIp(messages.getIp());
            List<Messages.Metric> metricsList = messages.getMetricsList();
            List<MetricEsDto> metrics = new ArrayList<>();
            if (!CollectionUtils.isEmpty(metricsList)) {
                metricsList.forEach(item -> {
                    MetricEsDto metricEsDto = new MetricEsDto();
                    metricEsDto.setMetricName(item.getMetricName());
                    metricEsDto.setMetricValue(item.getMetricVaule());
                    metrics.add(metricEsDto);
                });
            }
            esDto.setMetrics(metrics);
            IndexRequest indexRequest = new IndexRequest(gatherIndex, "_doc").source(JSONObject.toJSONString(esDto), XContentType.JSON);
            bulkIndexByIndexRequest(indexRequest);
        } catch (Exception e) {
            result = e.getMessage();
            log.error("============ 处理采集任务异常 errMsg:{} ===========", e.getMessage());
        }
        return result;
    }

    public void bulkIndexByIndexRequest(List<IndexRequest> index) {
        if (CollectionUtils.isEmpty(index)) {
            return;
        }
        BulkRequest request = new BulkRequest();
        index.stream().filter(Objects::nonNull).forEach(request::add);
        try {
            BulkResponse bulkResponse = elasticsearchRestTemplate.getClient().bulk(request, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                log.error("批量新增索引, failureMessage = {}", bulkResponse.buildFailureMessage());
            } else {
                log.info("批量新增索引成功,条数 {}", index.size());
            }
        } catch (IOException e) {
            log.error("批量新增索引失败, e ={}", e);
        }
    }

    public void bulkIndexByIndexRequest(IndexRequest index) {
        if (index != null) {
            BulkRequest request = new BulkRequest();
            request.add(index);
            try {
                BulkResponse bulkResponse = elasticsearchRestTemplate.getClient().bulk(request, RequestOptions.DEFAULT);
                if (bulkResponse.hasFailures()) {
                    log.error("索引, failureMessage = {}", bulkResponse.buildFailureMessage());
                }
            } catch (IOException e) {
                log.error("单条数据新增失败,index = {}, e ={}", 1, e);
            }
        }
    }


}
